package sundun.sharding.task;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import sundun.sharding.abnormal.entity.MigrationFail;
import sundun.sharding.abnormal.service.MigrationFailService;
import sundun.sharding.properties.MigrationTableConfiguration;
import sundun.sharding.properties.MigrationTableProperties;
import sundun.sharding.scaner.Invoker;
import sundun.sharding.scaner.InvokerManager;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.lang.reflect.InvocationTargetException;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;

/**
 * @Description:历史数据迁移至分片节点
 * @Author: xiaolong
 * @date: 2025/1/23
 * @Version: 1.0
 */
@ConditionalOnProperty(prefix = "sharding.migration", name = "enabled", havingValue = "true", matchIfMissing = true)
@EnableScheduling    //开启定时任务
@Component
public class DataMigrationTask {

    private static Logger logger = LogManager.getLogger(DataMigrationTask.class);


    private static final DateTimeFormatter START_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd 00:00:00");
    private static final DateTimeFormatter END_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd 23:59:59");

    private volatile boolean isRun = false;

    private static final String NUMBER_0 = "0";
    private static final String NUMBER_1 = "1";

    private static final String STARTTIME = "startTime";
    private static final String ENDTIME = "endTime";


    private  static final String MIGRATION_IMSTRAIL_MARK = "migration_%s_mark";


    @Value("${sharding.migration.thread.pool.core-size:6}")
    private int corePoolSize;

    @Value("${sharding.migration.thread.pool.max-size:12}")
    private int maxPoolSize;

    @Value("${sharding.migration.thread.pool.queue-capacity:2500}")
    private int queueCapacity;

    @Value("${sharding.migration.thread.pool.keep-alive:60}")
    private long keepAliveTime;

    private ExecutorService executorService;

    @Value("${sharding.migration.startTime:2023-12-27}")
    private String startDate;

    @Value("${sharding.migration.endTime:2025-01-23}")
    private String endDate;


    @Resource
    private RedisTemplate<String, Object> redisTemplate;


    @Autowired
    private MigrationTableProperties migrationTableProperties;

    @Autowired
    private MigrationFailService migrationFailService;

    @PostConstruct
    public void init() {
        executorService = new ThreadPoolExecutor(
                corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(queueCapacity),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @PreDestroy
    public void shutdown() {
        if (executorService != null && !executorService.isShutdown()) {
            executorService.shutdown();
        }
    }



    @Scheduled(cron = "${sharding.migration.cron: 0 0/2 * * * ?}")
    public void scan(){
        if(!isRun){
            isRun = true ;
            Instant startTime = Instant.now();
            LocalDate start = LocalDate.parse(startDate);
            LocalDate end = LocalDate.parse(endDate);
            Map<String, MigrationTableConfiguration> rule = migrationTableProperties.getRule();
            rule.forEach((key, value) -> {
                // 对每个 value 进行处理
                StartTask(start,end,key,value);
            });
            Instant endTime = Instant.now();
            long millis = Duration.between(startTime, endTime).toMillis();
            logger.info("数据迁移-执行完成,执行耗时={}s",millis/1000);
            isRun = false ;
        }
    }


    public void StartTask(LocalDate startDate, LocalDate endDate,String table,
                          MigrationTableConfiguration config) {
        List<CompletableFuture<Void>> futures = new ArrayList<>();

        if (!StringUtils.isEmpty(config.getSelectMethodMark()) && !StringUtils.isEmpty(config.getInsertMethodMark())) {
            Invoker selectInvoker = InvokerManager.getInvoker(config.getSelectMethodMark());
            Invoker insertInvoker = InvokerManager.getInvoker(config.getInsertMethodMark());
            if (!ObjectUtils.isEmpty(selectInvoker) && !ObjectUtils.isEmpty(insertInvoker)) {
                for (LocalDate date = startDate; !date.isAfter(endDate); date = date.plusDays(1)) {
                    LocalDate finalDate = date;
                    //判断是否重复执行
                    String startTime = finalDate.atStartOfDay().format(START_FORMATTER);
                    String format = String.format(MIGRATION_IMSTRAIL_MARK, table);
                    String mark = (String) redisTemplate.opsForHash().get(format, startTime);
                    if (StringUtils.isNotEmpty(mark) && !NUMBER_0.equals(mark)){
                        continue;
                    }
                    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                        try {
                            migrateDataForDate(finalDate,table,selectInvoker,insertInvoker,config);
                        } catch (Exception e) {
                            e.printStackTrace();
                            redisTemplate.opsForHash().put(format,startTime,NUMBER_0);
                        }
                    }, executorService);

                    futures.add(future);
                }
                // 等待所有任务完成
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            }
        }

    }

    public void migrateDataForDate(LocalDate date,String table,
                                   Invoker selectInvoker, Invoker insertInvoker,
                                   MigrationTableConfiguration config) throws InvocationTargetException, IllegalAccessException {
        String startTime = date.atStartOfDay().format(START_FORMATTER);
        String endTime = date.atStartOfDay().format(END_FORMATTER);
        String startTimeField = StringUtils.isNotEmpty(config.getStartTimeField()) ? config.getStartTimeField() : STARTTIME;
        String endTimeField = StringUtils.isNotEmpty(config.getEndTimeField()) ? config.getEndTimeField() : ENDTIME;
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put(startTimeField, startTime);
        paramMap.put(endTimeField, endTime);
        Object invoke = selectInvoker.invoke(paramMap);
        List<?> resultList = null;
        if (invoke instanceof List) {
            // 进行强制转换并处理结果
            resultList = (List<?>) invoke;
        }
        logger.info("【案卷数据迁移任务-开始】，开始日期=【{}】，结束日期=【{}】,数据量=【{}】,",startTime,endTime,resultList.size());
        String format = String.format(MIGRATION_IMSTRAIL_MARK, table);
        if (resultList.isEmpty()) {
            redisTemplate.opsForHash().put(format,startTime,NUMBER_1);
            return;
        }
        for (Object record : resultList) {
            try {
                insertInvoker.invoke(record);
            } catch (Exception e) {
                MigrationFail imsMigrationFail = new MigrationFail();
                if (e instanceof InvocationTargetException){
                    InvocationTargetException ex =  (InvocationTargetException) e;
                    imsMigrationFail.setException(ex.getTargetException().toString());
                } else {
                    imsMigrationFail.setException(stackTraceToString(e.getClass().getName(), e.getMessage(), e.getStackTrace()));
                }
                JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(record));
                imsMigrationFail.setId(getGuid32());
                imsMigrationFail.setFailNum(1);
                imsMigrationFail.setMigrationId(jsonObject.getString("id"));
                imsMigrationFail.setInfo(JSON.toJSONString(record));
                imsMigrationFail.setAddTime(new Date());
                imsMigrationFail.setUpdateTime(new Date());
                imsMigrationFail.setEndTime(endTime);
                imsMigrationFail.setStartTime(startTime);
                imsMigrationFail.setMigrationTable(table);
                migrationFailService.insert(imsMigrationFail);
            }
        }

        redisTemplate.opsForHash().put(format,startTime,NUMBER_1);
        logger.info("【案卷数据迁移任务-结束】，开始日期=【{}】，结束日期=【{}】",startTime,endTime,resultList.size());
    }


    public static String getGuid32() {
        return UUID.randomUUID().toString().replaceAll("-", "").toUpperCase();
    }



    /**
     * 转换异常信息为字符串
     *
     * @param exceptionName    异常名称
     * @param exceptionMessage 异常信息
     * @param elements         堆栈信息
     */
    public String stackTraceToString(String exceptionName, String exceptionMessage, StackTraceElement[] elements) {
        StringBuilder strbuff = new StringBuilder();
        for (StackTraceElement stet : elements) {
            strbuff.append(stet).append("\n");
        }
        return exceptionName + ":" + exceptionMessage + "\n\t" + strbuff.toString();
    }

}
